/* eslint-disable @typescript-eslint/no-unsafe-declaration-merging */
import { EventStreamMarshaller as UniversalEventStreamMarshaller } from "@smithy/eventstream-serde-universal";
import { Decoder, Encoder, EventStreamMarshaller as IEventStreamMarshaller, Message } from "@smithy/types";

import { readabletoIterable } from "./utils";
import { stream } from "@kit.ArkTS";

/**
 * @internal
 */
export interface EventStreamMarshaller extends IEventStreamMarshaller {}

/**
 * @internal
 */
export interface EventStreamMarshallerOptions {
  utf8Encoder: Encoder;
  utf8Decoder: Decoder;
}

/**
 * @internal
 */
export class EventStreamMarshaller {
  private readonly universalMarshaller: UniversalEventStreamMarshaller;
  constructor({ utf8Encoder, utf8Decoder }: EventStreamMarshallerOptions) {
    this.universalMarshaller = new UniversalEventStreamMarshaller({
      utf8Decoder,
      utf8Encoder,
    });
  }

  deserialize<T>(body: stream.Readable, deserializer: (input: Record<string, Message>) => Promise<T>): AsyncIterable<T> {
    //should use stream[Symbol.asyncIterable] when the api is stable
    //reference: https://nodejs.org/docs/latest-v11.x/api/stream.html#stream_readable_symbol_asynciterator
    const bodyIterable: AsyncIterable<Uint8Array> =readabletoIterable(body);
    return this.universalMarshaller.deserialize(bodyIterable, deserializer);
  }

  serialize<T>(input: AsyncIterable<T>, serializer: (event: T) => Message): stream.Readable {
    // return Readable.from(this.universalMarshaller.serialize(input, serializer));
    return new ReadableStream(this.universalMarshaller.serialize(input, serializer))
  }

}

class ReadableStream extends stream.Readable{
  private iterator: AsyncIterator<Uint8Array>;
  constructor(asyncIter:AsyncIterable<Uint8Array>) {
    super();
    this.iterator = asyncIter[Symbol.asyncIterator]();
  }

  async doRead(size: number): Promise<void> {
    const { value, done } = await this.iterator.next();
    if (done) {
      this.push(null);
    } else {
      this.push(value);
    }
  }
}